﻿using Adaptive.Agrona;
using System;

using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using Unity.Commons;
using Unity.Connections;
using Unity.Credentials;
using Unity.Logger;

namespace Unity.Kafka
{
    public class UnityKafkaProducer : IUnityProducer
    {
        private static readonly ILogger Logger = UnityLogger.ConsoleLoggerFactory.CreateLogger(nameof(UnityKafkaProducer));

        private readonly UnityKafkaCredentials _credentials;
        private IProducer<int, byte[]> _producer;
        private int _counter;

        public UnityKafkaProducer(UnityKafkaCredentials credentials)
        {
            _credentials = credentials;
            _counter = 0;
        }


        public bool Connect(IUnityCredentials credentials)
        {
            try
            {
                _producer = _credentials.GetProducer();
                return Produce(Array.Empty<byte>());
            }
            catch(Exception exception)
            {
                Logger.LogError("Error connect to Kafka producer. Exception: {}", exception.Message);
                return false;
            }
        }

        public void Close()
        {
            _producer?.Dispose();
        }

        public void Dispose()
        {
            Close();
        }

        public bool Produce(byte[] data)
        {
            try
            {
                _producer.Produce(_credentials.Topic,
                    new Message<int, byte[]>
                    {
                        Key = _counter++,
                        Value = data
                    });
                return true;
            }
            catch(Exception exception)
            {
                Logger.LogError("Error produce to Kafka producer. Exception: {}", exception.Message);
                return false;
            }
        }

        public bool Produce(IDirectBuffer buffer, int offset, int length)
        {
            byte[] data = new byte[length];
            buffer.GetBytes(0, data, offset, length);
            return Produce(data);
        }

        public bool AddToBatching(IDirectBuffer buffer, int offset, int length)
        {
            return Produce(buffer, offset, length);
        }

        public int CompleteBatching()
        {
            return _counter;
        }

    }
}
